package InitialRank;

import java.io.IOException;
import java.util.Arrays;
import java.util.Iterator;

import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import removeDanglingPage.ReadBytes;

import mapred.AbstractReducer;
import mapred.util.Bytes;

public class InitialRank_Reducer extends AbstractReducer<BytesWritable, BytesWritable, BytesWritable, BytesWritable>{

	double hostrank;
	double pagerank;
	byte[] host;
	@Override
	public void reduce(BytesWritable key, Iterator<BytesWritable> values,
			OutputCollector<BytesWritable, BytesWritable> output,
			Reporter reporter) throws IOException {
		// TODO Auto-generated method stub
		ReadBytes k = new ReadBytes(key);
		ReadBytes v = new ReadBytes(values.next());
		double rank = v.nextDouble();
		double initialRank = 1d;
		if(k.hasNextUrl()) {
			byte[] url = k.nextUrl();
			pagerank = rank;
			if(Arrays.equals(host, Arrays.copyOf(url, 8))) {
				initialRank = pagerank * hostrank;
				output.collect(key, new BytesWritable(Bytes.Double2Bytes(initialRank)));
			} else {
				output.collect(key, new BytesWritable(Bytes.Double2Bytes(pagerank)));
			}
		}
		if(k.hasNextBytes(8)) {
			host = k.nextBytes(8);
			hostrank = rank;
		}
	}

}
